package com.galeno.day03.transfer;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.nio.charset.StandardCharsets;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/22 15:06
 * <p>
 * 实时统计广告点击的人数和次数
 * #广告id,用户id
 * ad1,user1
 * ad1,user1
 * ad1,user2
 * ad2,user1
 * ad2,user2
 * <p>
 * #结果
 * ad1,2,3
 * ad2,2,2
 */
public class AdUVPV {
    public static void main(String[] args) throws Exception {
        //BloomFilter<Integer> integerBloomFilter = BloomFilter.create(Funnels.integerFunnel(), Integer.MAX_VALUE);


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 22222);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.enableCheckpointing(1000);
        DataStreamSource<String> lines = env.socketTextStream("galeno", 9999);
        SingleOutputStreamOperator<Tuple4<String, String, Integer, Integer>> tp4 = lines.map(x -> {
            String[] split = x.split("\\s+");
            String ad = split[0];
            String user = split[1];
            if (ad.contains("error")) {
                throw new RuntimeException("计算出现异常,等待重试");
            }
            return Tuple4.of(ad, user, 1, 1);
        }).returns(TypeInformation.of(new TypeHint<Tuple4<String, String, Integer, Integer>>() {
        }));
        KeyedStream<Tuple4<String, String, Integer, Integer>, String> tp4keyBy = tp4.keyBy(x -> x.f0);
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> map = tp4keyBy.map(new RichMapFunction<Tuple4<String, String, Integer, Integer>, Tuple3<String, Integer, Integer>>() {
            private ValueState<Integer> valueStateUV;
            private ValueState<Integer> valueStatePV;
            private ValueState<BloomFilter<String>> valueStateBloomFilter;


            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Integer> valueStateDescriptorUV = new ValueStateDescriptor<>("UV-Count-state", Integer.class);
                ValueStateDescriptor<Integer> valueStateDescriptorPV = new ValueStateDescriptor<>("PV-Count-state", Integer.class);
                ValueStateDescriptor<BloomFilter<String>> valueBloomFilterStateDescriptor = new ValueStateDescriptor<>("BloomFilter-state",
                        TypeInformation.of(new TypeHint<BloomFilter<String>>() {
                        }));
                valueStateUV = getRuntimeContext().getState(valueStateDescriptorUV);
                valueStatePV = getRuntimeContext().getState(valueStateDescriptorPV);
                valueStateBloomFilter = getRuntimeContext().getState(valueBloomFilterStateDescriptor);
            }

            @Override
            public Tuple3<String, Integer, Integer> map(Tuple4<String, String, Integer, Integer> in) throws Exception {
                System.out.println("-------------------------------------------------");
                System.out.println(valueStateUV.value());
                System.out.println(valueStatePV.value());
                System.out.println(valueStateBloomFilter.value());
                System.out.println("-------------------------------------------------");

                // bloomFilter = BloomFilter.create(Funnels.stringFunnel(StandardCharsets.UTF_8), Integer.MAX_VALUE, 0.00001);
                String  user = in.f1;
                Integer PVCount = in.f2;
                Integer UVCount = in.f3;
                Integer historyPVCount = valueStatePV.value();
                Integer historyUVCount = valueStateUV.value();
                BloomFilter<String> bloomFilter = valueStateBloomFilter.value();
                if (historyPVCount == null) {
                    historyPVCount = 0;
                }
                int totalPVCount = historyPVCount + PVCount;
                valueStatePV.update(totalPVCount);
                if (historyUVCount == null) {
                    historyUVCount = 0;
                }
                if (bloomFilter == null) {
                    bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 10000000, 0.0001);
                }
                if (!bloomFilter.mightContain(user)) {
                    int totalUVCount = historyUVCount + UVCount;
                    System.out.println("用户不存在,映射到布隆过滤器中");
                    bloomFilter.put(user);
                    valueStateUV.update(totalUVCount);
                    valueStateBloomFilter.update(bloomFilter);
                    return Tuple3.of(in.f0, totalPVCount, totalUVCount);
                } else
                    System.out.println("存在该用户");
                //valueStateBloomFilter.update(bloomFilter);
                return Tuple3.of(in.f0, totalPVCount, historyUVCount);
            }
        });

        map.print();
        env.execute();
    }
}
